package cn.monkey.transport.mq.redisson;

import org.redisson.api.*;
import org.redisson.client.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;

public class SimpleRedissonInbound implements RedissonInbound {
    protected final String fromTopic;
    protected final RedissonClient redissonClient;
    protected final RBlockingQueue<Object> blockQueue;
    protected final Codec codec;
    protected final int pollSize;
    protected final int pollFrequencyMs;
    protected final AtomicBoolean closed;
    protected final Flux<Object> receiver;

    public SimpleRedissonInbound(String fromTopic,
                                 RedissonClient redissonClient,
                                 Codec codec,
                                 int pollSize,
                                 int pollFrequencyMs) {
        this.fromTopic = fromTopic;
        this.pollSize = pollSize;
        this.pollFrequencyMs = pollFrequencyMs;
        this.codec = codec;
        this.redissonClient = redissonClient;
        this.closed = new AtomicBoolean(false);
        this.blockQueue = this.codec == null ? redissonClient.getBlockingQueue(fromTopic) : redissonClient.getBlockingQueue(fromTopic, codec);
        Object lock = new Object();
        this.receiver = Flux.create(sink -> {
            while (!this.closed.get()) {
                CompletableFuture<List<Object>> completableFuture = this.blockQueue.pollAsync(this.pollSize())
                        .toCompletableFuture();
                completableFuture.thenAccept(objects -> {
                    if (objects == null || objects.isEmpty()) {
                        synchronized (lock) {
                            try {
                                lock.wait(this.pollFrequencyMs());
                            } catch (InterruptedException e) {
                                sink.error(e);
                            }
                        }
                        return;
                    }
                    for (Object o : objects) {
                        sink.next(o);
                    }
                    synchronized (lock) {
                        lock.notifyAll();
                    }
                });
                synchronized (lock) {
                    try {
                        lock.wait(this.pollFrequencyMs());
                    } catch (InterruptedException e) {
                        sink.error(e);
                        return;
                    }
                }
            }
            sink.complete();
        });
    }

    @Override
    public Codec codec() {
        return this.codec;
    }

    @Override
    public RedissonClient redissonClient() {
        return this.redissonClient;
    }

    @Override
    public int pollSize() {
        return this.pollSize;
    }

    @Override
    public int pollFrequencyMs() {
        return this.pollFrequencyMs;
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver.filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }

    @Override
    public String topic() {
        return this.fromTopic;
    }

    @Override
    public Mono<Void> close() {
        this.closed.set(true);
        RFuture<Boolean> future = this.blockQueue.deleteAsync();
        return Mono.fromCompletionStage(future).then();
    }
}
